package yuekao3.util;

import com.alibaba.fastjson.JSON;
import org.apache.flink.api.java.tuple.*;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import yuekao3.entity.*;

import java.sql.*;
import java.sql.DriverManager;
import java.util.Collections;

public class AsyncIOUtil extends RichAsyncFunction<Tuple2<OrderInfo, OrderDetail>,Tuple5<OrderInfo,OrderDetail, UserInfo, BaseProvince,SkuInfo>> {
    Connection conn = null;
    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        conn = DriverManager.getConnection("jdbc:phoenix:hadoop-single:2181", "", "");
        conn.setAutoCommit(true);
    }

    @Override
    public void close() throws Exception {
        conn.close();
    }

    @Override
    public void asyncInvoke(Tuple2<OrderInfo, OrderDetail> value, ResultFuture<Tuple5<OrderInfo, OrderDetail, UserInfo, BaseProvince,SkuInfo>> resultFuture) throws Exception {
        BaseProvince baseprovince=null;
        PreparedStatement pds = conn.prepareStatement("select * from dim_base_province where id=?");
        pds.setObject(1,value.f0.getProvince_id());
        ResultSet rs = pds.executeQuery();
        while (rs.next()){
            String line = rs.getString("line");
            String data = JSON.parseObject(line).getString("data");
            baseprovince=JSON.parseObject(data,BaseProvince.class);
        }

        UserInfo userinfo=null;
        PreparedStatement pds1 = conn.prepareStatement("select * from dim_user_info where id=?");
        pds1.setObject(1,value.f0.getUser_id());
        ResultSet rs1 = pds1.executeQuery();
        while (rs1.next()){
            String line = rs1.getString("line");
            String data = JSON.parseObject(line).getString("data");
            userinfo=JSON.parseObject(data,UserInfo.class);
        }

        SkuInfo skuinfo=null;
        PreparedStatement pds2 = conn.prepareStatement("select * from dim_sku_info where id=?");
        pds2.setObject(1,value.f1.getSku_id());
        ResultSet rs2 = pds2.executeQuery();
        while (rs2.next()){
            String line = rs2.getString("line");
            String data = JSON.parseObject(line).getString("data");
            skuinfo=JSON.parseObject(data,SkuInfo.class);
        }

        resultFuture.complete(Collections.singleton((new Tuple5<>(value.f0,value.f1,userinfo,baseprovince,skuinfo))));
    }
       
}
